package com.sun.rabbit.workQueues.auto_ack;


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;
import com.sun.rabbit.utils.RabbitMqUtils;

import java.io.IOException;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author shkstart
 * @create 2021-10-14 16:47
 */
public class Task02 {
    private final static String TASk_QUEUE_NAME="ack_queue";
    private final static int MESSAGE_COUNT=1000;//消息总数

    public static void main(String[] args) throws Exception {
        //简单发布确认： 发布1000 个单独确认消息, 耗时7199ms
        //publishMessageIndividually();
        //批量发布确认:发布1000 个批量确认消息, 耗时260ms
        //publishMessageBatch();// 发布1000 个批量确认消息, 耗时190ms
        // publishMessageBatch1();
        //异步发布确认:    发布1000 个异步确认消息, 耗时131ms
        publishMessageAsync();
    }
    //异步发布确认：
    public static void publishMessageAsync() throws Exception{
        try(Channel channel=RabbitMqUtils.getChannel();){
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName,false,false,false,null);
            //开启发布确认模式
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1. 轻松的将序号与消息进行关联
             * 2. 轻松批量删除条目 只要给到序列号
             * 3. 支持并发访问
             */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<Long, String>();

            /**
             * 确认收到消息的一个回调
             * 1. 消息序列号
             * 2.true 可以确认小于等于当前序列号的消息
             * false 确认当前序列号消息
             */
            //消息确认成功 回调函数
            ConfirmCallback ackCallback=(deliveryTag, multiple)->{

                if(multiple){
                    //批量删除
                    //2.删除已经确认的消息，剩下的就是未确认的消息
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag,true);
                    confirmed.clear();
                }else{
                    outstandingConfirms.remove(deliveryTag);
                }
                System.out.println("确认的消息:"+deliveryTag);

            };
            //消息确认失败 回调函数
            /**
             * 1.消息的标识
             * 2.是否为批量确认
             */
            ConfirmCallback nackCallback=(deliveryTag,multiple)->{
                //3.打印一下未确认的消息有哪些
                String message = outstandingConfirms.get(deliveryTag);
                // TODO: 2021/10/16  4.重新发送，待处理

                System.out.println(" 发布的消息"+message+" 未被确认，序列号"+deliveryTag);
            };
            /**
             * 添加一个异步确认的监听器
             * 1. 监听那些消息成功
             * 2. 哪些消息失败
             */
            //消息监听:准备消息监听器，监听那些消息成功了，哪些消息失败了 异步通知
            channel.addConfirmListener(ackCallback,nackCallback);
            long begin=System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message=i+" ";
                /**
                 * channel.getNextPublishSeqNo() 获取下一个消息的序列号
                 * 通过序列号与消息体进行一个关联
                 * 全部都是未确认的消息体
                 */
                channel.basicPublish("",queueName,null,message.getBytes());
                //1.记录下所有要发送的消息，消息总和
                outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
            }
            long end=System.currentTimeMillis();
            System.out.println(" 发布" + MESSAGE_COUNT + " 个异步确认消息, 耗时" + (end - begin) + "ms");
        }
    }

    //批量的发布确认：当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种
    //方案仍然是同步的，也一样阻塞消息的发布
    public static void publishMessageBatch1() throws Exception{
        try(Channel channel=RabbitMqUtils.getChannel();){
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName,true,false,false,null);
            //开启发布确认模式
            channel.confirmSelect();
            long begin=System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message=i+" ";
                channel.basicPublish("",queueName,null,message.getBytes("UTF-8"));
                if(i%100==0){
                   channel.waitForConfirms();
               }
            }
            long end = System.currentTimeMillis();
            System.out.println(" 发布" + MESSAGE_COUNT + " 个批量确认消息, 耗时" + (end - begin) + "ms");
        }
    }

    public static void publishMessageBatch() throws Exception{
        try(Channel channel=RabbitMqUtils.getChannel();){
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName,true,false,false,null);
            //开启发布确认模式
            channel.confirmSelect();
            //定义批量确认消息的大小
            int batchSize=100;
            //未确认消息的个数
            int outstandingMessageCount=0;
            long begin=System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message=i+" ";
                channel.basicPublish("",queueName,null,message.getBytes("UTF-8"));
                outstandingMessageCount++;
                if(outstandingMessageCount==batchSize){
                    channel.waitForConfirms();
                    outstandingMessageCount=0;
                }
            }
            //为了确保还有剩余没有确认消息 再次确认
            if (outstandingMessageCount > 0) {
                channel.waitForConfirms();
            }
            long end = System.currentTimeMillis();
            System.out.println(" 发布" + MESSAGE_COUNT + " 个批量确认消息, 耗时" + (end - begin) + "ms");

        }
    }

    //简单的发布确认：保证消息传入对列过程中，不丢失(同步确认发布)发布速度特别的慢
    private static void publishMessageIndividually() throws Exception{

        try(Channel channel = RabbitMqUtils.getChannel();){
            //定义随机对列名称
            String queueName= UUID.randomUUID().toString();
            System.out.println(queueName);
            /**
             * 生成一个队列
             * 1. 队列名称
             * 2. 队列里面的消息是否持久化 默认消息存储在内存中
             * 3. 该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
             * 4. 是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除
             * 5. 其他参数
             */
            channel.queueDeclare(queueName,true,false,false,null);
            //开启发布确认
            channel.confirmSelect();
            long begin=System.currentTimeMillis();
            for(int i=0;i<MESSAGE_COUNT;i++){
                String message=i+" ";
                channel.basicPublish("",queueName,null,message.getBytes("UTF-8"));
                /**
                 * 发布一个消息之后只有它
                 * 被确认发布，后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认
                 * 的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常
                 */
                // 服务端返回 false 或超时时间内未返回，生产者可以消息重发
                boolean flag = channel.waitForConfirms();
                if (flag){
                    System.out.println("消息发送成功！");
                }
            }
            long end=System.currentTimeMillis();
            System.out.println(" 发布" + MESSAGE_COUNT + " 个单独确认消息, 耗时" + (end - begin) + "ms");
        }
    }
    //队列和消息持久化
    private static void common() throws Exception{
        try (Channel channel = RabbitMqUtils.getChannel();){

            //channel.queueDeclare(TASk_QUEUE_NAME,false,false,false,null);
            //让消息队列持久化
            boolean durable=true;
            channel.queueDeclare(TASk_QUEUE_NAME,durable,false,false,null);
            Scanner scanner = new Scanner(System.in);
            System.out.println("请输入消息:");
            while (scanner.hasNext()){
                String message = scanner.nextLine();
                //channel.basicPublish("",TASk_QUEUE_NAME,null,message.getBytes("UTF-8"));
                //消息实现持久化
                channel.basicPublish("",TASk_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
                System.out.println("发送消息完成"+message);
            }
        }
    }
}
